package org.yumeko.kafka.Producer;

import org.apache.kafka.clients.producer.ProducerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RestController;

import java.util.concurrent.CompletableFuture;

@RestController
public class KafkaAsyncProducerController {
    @Autowired
    private KafkaTemplate<Integer, String> kafkaTemplate;

    @GetMapping("send/async")
    public String asyncSend(@RequestParam("message") String message) {
        CompletableFuture<SendResult<Integer, String>> future = kafkaTemplate.send(
                new ProducerRecord<>(
                        "topic-spring",
                        0,
                        3,
                        "[异步发送] " + message
                ));
        // 添加回调，异步等待响应
        future.whenComplete((result, throwable) -> {
            if (throwable != null) {
                System.err.println("[异步] 发送失败: " + throwable.getMessage());
                throwable.printStackTrace();
            } else {
                System.out.println("[异步] 发送成功：" +
                        result.getRecordMetadata().topic() + "\t"
                        + result.getRecordMetadata().partition() + "\t"
                        + result.getRecordMetadata().offset());
            }
        });

        // 立即返回成功响应
        return "success";
    }
}